package com.nexgo.data.net;

import android.text.TextUtils;

import com.koushikdutta.async.AsyncServer;
import com.koushikdutta.async.Util;
import com.koushikdutta.async.callback.ConnectCallback;
import com.koushikdutta.async.future.Cancellable;
import com.nexgo.common.ByteUtils;
import com.nexgo.data.exception.RequestException;
import com.nexgo.data.exception.ResponseException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.security.InvalidParameterException;
import java.util.Arrays;

import io.reactivex.Observable;

/**
 * Created by xiaox on 2017/6/2.
 */

public class SocketComm {
    private Logger log = LoggerFactory.getLogger(SocketComm.class.getSimpleName());

    private String mHost;
    private int mPort;
    private int mTimeout;
    private Cancellable mCancellable;
    private Buffer mBuffer;
    private boolean mIsWrited;
    private boolean mIsCompleted;

    public SocketComm() {
    }

    public SocketComm(String host, int port, int timeout) {
        mHost = host;
        mPort = port;
        mTimeout = timeout;
    }

    public String getHost() {
        return mHost;
    }

    public void setHost(String host) {
        this.mHost = host;
    }

    public int getPort() {
        return mPort;
    }

    public void setPort(int port) {
        this.mPort = port;
    }

    public int getTimeout() {
        return mTimeout;
    }

    public void setTimeout(int timeout) {
        this.mTimeout = timeout;
    }

    public Observable<byte[]> sendAndRecv(final byte[] request) {
        mIsCompleted = false;
        mIsWrited = false;
        return Observable.create(emitter -> {
            if (TextUtils.isEmpty(mHost)) {
                emitter.onError(new InvalidParameterException());
                return;
            }
            log.debug("ip {},port {},timeout {}", mHost, mPort, mTimeout);
            mBuffer = new Buffer();
            log.debug("开启定时器");
            AsyncServer.getDefault().postDelayed(() -> {
                if(!emitter.isDisposed()) {
                    if (!mIsCompleted) {
                        log.debug("超时，取消网络连接");
                        mCancellable.cancel();
                        if (mIsWrited) {
                            log.debug("此时数据已发送");
                            emitter.onError(new ResponseException("此时数据已发送"));
                        } else {
                            log.debug("此时数据未发送");
                            emitter.onError(new RequestException("此时数据未发送"));
                        }
                    }
                }
            }, mTimeout * 1000);
            ConnectCallback callback = (ex, socket) -> {
                if (ex != null) {
                    ex.printStackTrace();
                    log.debug("网络连接失败");
                    emitter.onError(new RequestException("网络连接失败"));
                    return;
                }
                Util.writeAll(socket, request, ex1 -> {
                    if (ex1 != null) {
                        log.debug("发送数据失败");
                        emitter.onError(new RequestException("发送数据失败"));
                        return;
                    }
                    log.debug("发送数据成功");
                    mIsWrited = true;
                });
                socket.setDataCallback((dataEmitter, bb) -> {
                    if (bb == null) return;
                    byte[] recv = bb.getAllByteArray();
                    log.debug("收到socket数据\n{}", ByteUtils.byteArray2HexString(recv));
                    mBuffer.write(recv);
                    bb.recycle();
                    byte[] allByteArray = mBuffer.toByteArray();
                    //两个字节长度
                    byte[] lis = Arrays.copyOfRange(allByteArray, 0, 2);
                    log.debug("长度指标位 {}", ByteUtils.byteArray2HexString(lis));
                    int len = ((lis[0] & 0xFF) << 8) + (lis[1] & 0xFF);
                    log.debug("期望后续长度 {},收到本次总长度 {}", len, allByteArray.length);
                    if (len == allByteArray.length - 2) {
                        socket.close();
                        emitter.onNext(mBuffer.toByteArray());
                        emitter.onComplete();
                        mIsCompleted = true;
                    } else {
                        log.debug("等待后续数据");
                    }
                });
            };
            mCancellable = AsyncServer.getDefault().connectSocket(mHost, mPort, callback);
        });
    }

    private class Buffer {

        private int count;
        private byte[] buf;

        Buffer() {
            buf = new byte[32];
        }

        private void expand(int i) {
            if (count + i <= buf.length) {
                return;
            }

            byte[] newbuf = new byte[(count + i) * 2];
            System.arraycopy(buf, 0, newbuf, 0, count);
            buf = newbuf;
        }

        void backspace() {
            if (count > 0) count--;
        }

        void reset() {
            count = 0;
        }

        void write(byte oneByte) {
            if (count == buf.length) {
                expand(1);
            }
            buf[count++] = oneByte;
        }

        void write(byte[] bytes) {
            for (byte oneByte : bytes) {
                write(oneByte);
            }
        }

        byte[] toByteArray() {
            byte[] newArray = new byte[count];
            System.arraycopy(buf, 0, newArray, 0, count);
            return newArray;
        }
    }
}
